<?PHP
namespace App\Http\Services;
use Kafka;

use Monolog\Logger;
//use Monolog\Handler\StdoutHandler;

class KafkaService
{
    public function __construct()
    {
        date_default_timezone_set('PRC');
    }

    public function Producer($topic,$value,$url){

// Create the logger
        $logger = new Logger('my_logger');


        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList($url);
        $config->setBrokerVersion('1.0.0');
        $config->setRequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);
        $producer = new \Kafka\Producer(function() use($topic,$value) {
            return array(
                array(
                    'topic' => $topic,
                    'value' => 'test....message.',
                    'key' => 'testkey',
                ),
            );
        });
        //$producer->setLogger($logger);
        $producer->success(function($result) {
            var_dump($result);
        });
        $producer->error(function($errorCode) {
            var_dump($errorCode);
        });
        $producer->send(true);
    }
    /*
     * Produce
     */
    public function Producer2($topic,$value,$url)
    {
        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadatabrokerList($url);
        $config->setbrokerVersion('2.8.1');
        $config->setrequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);

       // echo 'topic:'.$topic.':value:'.$value;
        $producer = new \Kafka\Producer(function() use($topic,$value){
            return array(
                array(
                    'topic' => $topic,
                    'value' => $value,
                    'key' => 'testkey',
                ),
            );
        });

        $producer->success(function ($result){
            var_dump($result);
            return "success";
        });
        $producer->error(function ($errorCode){
            echo PHP_EOL.'--errorCode---'.PHP_EOL;
            echo $errorCode;
            //var_dump($errorCode);
        });
        $producer->send(true);
    }
    /*
     * Consumer
     */
    public function consumer($group,$topics,$url){
        $config = \Kafka\ConsumerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList($url);
        $config->setGroupId($group);
        $config->setbrokerVersion('1.0.0');
        $config->setTopics([$topics]);
        $config->setoffsetReset('earliest');
        $consumer = new \Kafka\Consumer();
       // echo '-----'.$topics. '-----';

        $consumer->start(function($topics,$part,$message) {
            echo '--1-';
            echo $topics;
            var_dump($message);
            echo "receive a message...\n";
        });
    }
}
